package net.bwie.douyin.dwd.log.mock;

import com.google.gson.Gson;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;

public class LiveEventProducer {
    private static final String TOPIC = "live_events";
    private static final String BOOTSTRAP_SERVERS = "node101:9092";

    public static void main(String[] args) {
        // Kafka配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            generateAndSendEvents(producer); // 生成并发送数据
        }
    }


    private static void generateAndSendEvents(KafkaProducer<String, String> producer) {
        Gson gson = new Gson();
        Random random = new Random();
        LocalDateTime baseTime = LocalDateTime.of(2025, 5, 30, 0, 0); // 基准时间

        // 生成10个直播间及对应的主播
        Map<String, String> liveToAnchor = new HashMap<>();
        for (int i = 1; i <= 10; i++) {
            liveToAnchor.put("live_" + i, "anchor_" + i);
        }

        // 每个用户访问1~3个直播间
        for (int userId = 1; userId <= 10000; userId++) {
            String userIdStr = "user_" + userId;
            List<String> liveIds = selectRandomLives(liveToAnchor.keySet(), 1 + random.nextInt(3));

            for (String liveId : liveIds) {
                // 生成进入事件
                LiveEvent enterEvent = buildEnterEvent(userIdStr, liveId, liveToAnchor.get(liveId), baseTime);
                sendToKafka(producer, gson.toJson(enterEvent));

                // 生成中间事件（1~5次）
                int actions = 1 + random.nextInt(5);
                long currentTs = enterEvent.ts;
                for (int i = 0; i < actions; i++) {
                    currentTs += 1000 * (1 + random.nextInt(60)); // 间隔1~60秒
                    LiveEvent actionEvent = generateRandomAction(userIdStr, liveId, currentTs, random);
                    sendToKafka(producer, gson.toJson(actionEvent));
                }

                // 生成离开事件
                currentTs += 1000 * (1 + random.nextInt(60));
                LiveEvent leaveEvent = buildLeaveEvent(userIdStr, liveId, liveToAnchor.get(liveId), currentTs);
                sendToKafka(producer, gson.toJson(leaveEvent));
            }
        }
    }

    // 辅助方法：随机选择直播间
    private static List<String> selectRandomLives(Set<String> lives, int count) {
        List<String> liveList = new ArrayList<>(lives);
        Collections.shuffle(liveList);
        return liveList.subList(0, Math.min(count, liveList.size()));
    }

    private static LiveEvent buildLeaveEvent(String userId, String liveId, String anchorId, long ts) {
        LiveEvent event = new LiveEvent();
        event.eventType = "leave";  // 事件类型必须为 leave
        event.userId = userId;
        event.liveId = liveId;
        event.roomId = liveId;      // 房间ID与直播ID一致
        event.ts = ts;              // 传入的时间戳（需在外部计算）
        event.anchorId = anchorId;
        return event;
    }

    // 辅助方法：构建进入事件
    private static LiveEvent buildEnterEvent(String userId, String liveId, String anchorId, LocalDateTime baseTime) {
        LiveEvent event = new LiveEvent();
        event.eventType = "enter";
        event.userId = userId;
        event.liveId = liveId;
        event.anchorId = anchorId;
        event.roomId = liveId;
        event.ts = baseTime.toEpochSecond(ZoneOffset.UTC) * 1000; // 转毫秒
        return event;
    }

    // 辅助方法：生成随机动作事件
    private static LiveEvent generateRandomAction(String userId, String liveId, long ts, Random random) {
        LiveEvent event = new LiveEvent();
        event.userId = userId;
        event.liveId = liveId;
        event.roomId = liveId;
        event.ts = ts;

        String[] types = {"gift", "message", "share", "like"};
        String type = types[random.nextInt(types.length)];
        event.eventType = type;

        switch (type) {
            case "gift":
                event.giftId = 1 + random.nextInt(10);
                event.giftValue = 10 + random.nextInt(991); // 10~1000
                break;
            case "message":
                event.messageContent = generateRandomMessage(random);
                break;
            // share/like无需额外字段
        }
        return event;
    }

    // 辅助方法：生成随机消息内容
    private static String generateRandomMessage(Random random) {
        String[] words = {"精彩!", "666", "关注了", "求礼物", "主播加油", "再来一个"};
        return String.join(" ",
                words[random.nextInt(words.length)],
                words[random.nextInt(words.length)],
                words[random.nextInt(words.length)]
        );
    }

    private static void sendToKafka(KafkaProducer<String, String> producer, String json) {
        producer.send(new ProducerRecord<>(TOPIC, json));
        // 可选：添加回调处理异常
    /*.addCallback(result -> {
        // 发送成功逻辑
    }, ex -> {
        System.err.println("发送失败: " + ex.getMessage());
    });*/
    }
}

class LiveEvent {
    String eventId = UUID.randomUUID().toString(); // 唯一事件ID
    String liveId;     // 直播ID（1~1000）
    String anchorId;   // 主播ID（与直播ID绑定）
    String userId;     // 用户ID（1~10000）
    String roomId;     // 房间ID（与直播ID相同）
    long ts;           // 时间戳（2025-05-27当天）
    String eventType;  // enter/gift/message/share/like/leave
    Integer giftId;    // 礼物ID（1~10）
    Integer giftValue; // 礼物价值（10~1000）
    String messageContent; // 随机消息

    // 构造函数、Getter/Setter省略（需根据eventType初始化不同字段）
}

